package com.alibaba.nacos.core.remote.grpc;

import com.alibaba.nacos.api.grpc.auto.BiRequestStreamGrpc;
import com.alibaba.nacos.api.grpc.auto.Payload;
import com.alibaba.nacos.api.grpc.auto.RequestGrpc;
import com.alibaba.nacos.api.naming.remote.request.InstanceRequest;
import com.alibaba.nacos.api.naming.remote.response.InstanceResponse;
import com.alibaba.nacos.api.remote.request.ConnectionSetupRequest;
import com.alibaba.nacos.api.remote.request.ServerCheckRequest;
import com.alibaba.nacos.api.remote.response.ServerCheckResponse;
import com.alibaba.nacos.common.remote.client.grpc.GrpcUtils;
import com.google.common.util.concurrent.ListenableFuture;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

import static com.alibaba.nacos.common.utils.JacksonUtils.toJson;

@Slf4j
class GrpcBiStreamRequestAcceptorTest3 {
    int port = 9848;
//    int port = 9858;
    ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", port).usePlaintext().build();
    BiRequestStreamGrpc.BiRequestStreamStub biRequestStreamStub = BiRequestStreamGrpc.newStub(channel);
    RequestGrpc.RequestFutureStub requestFutureStub = RequestGrpc.newFutureStub(channel);

    @SneakyThrows
    @Test
    void testConnectionSetupRequest() {
        ServerCheckRequest serverCheckRequest = RequestTestUtils.getServerCheckRequest();
        Payload request = GrpcUtils.convert(serverCheckRequest);
        ListenableFuture<Payload> response = requestFutureStub.request(request);
        ServerCheckResponse serverCheckResponse  = (ServerCheckResponse) GrpcUtils.parse(response.get());
        log.info("\nserverCheckRequest:{}\nserverCheckResponse:{}", toJson(serverCheckRequest), toJson(serverCheckResponse));


        ConnectionSetupRequest connectionSetupRequest = RequestTestUtils.getConnectionSetupRequest();
        request = GrpcUtils.convert(connectionSetupRequest);
        StreamObserver<Payload> payloadStreamObserver = bindRequestStream(biRequestStreamStub);
        payloadStreamObserver.onNext(request);
        LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(2));


        InstanceRequest instanceRequest = RequestTestUtils.getInstanceRequest();
        request = GrpcUtils.convert(instanceRequest);
        response = requestFutureStub.request(request);
        InstanceResponse instanceResponse = (InstanceResponse) GrpcUtils.parse(response.get());
        log.info("\ninstanceResponse:{}\n", toJson(instanceResponse));


        System.in.read();
    }

    private StreamObserver<Payload> bindRequestStream(BiRequestStreamGrpc.BiRequestStreamStub biRequestStreamStub) {
        return biRequestStreamStub.requestBiStream(new StreamObserver<Payload>() {
            @Override
            public void onNext(Payload value) {
                log.info("onNext value:{}", toJson(value));
            }

            @Override
            public void onError(Throwable t) {
                log.error(t.getMessage(), t);
            }

            @Override
            public void onCompleted() {
                log.info("onCompleted");
            }
        });
    }


}